package streaming.api.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import utils.PropertiesReader;

/**
 * 从文件获取数据 (SQL-DDL)
 * 聚合统计
 * 输出到控制台展示
 */
public class TableTest3 {

    private static String filePath = PropertiesReader.get("default.file.from.path");

    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 2. 表的创建：连接外部系统，读取数据
        String fileFormat = "csv";

        StringBuffer sb1 = new StringBuffer();
        sb1.append("CREATE TABLE inputTable (");
        sb1.append("id STRING, `timestamp` BIGINT, temp DOUBLE");
        sb1.append(") WITH (");
        sb1.append("'connector.type' = 'filesystem', ");
        sb1.append("'connector.path' = '").append(filePath).append("', ");
        sb1.append("'format.type' = '").append(fileFormat).append("'");
        sb1.append(")");
        System.out.println("inputTable SQL: " + sb1.toString());
        tableEnv.executeSql(sb1.toString());
        // 转换
        Table inputTable = tableEnv.from("inputTable");

        // 3.1 简单转换
        Table resultTable = inputTable.select("id, temp").filter("id === 'sensor_c'");
        // 打印输出
        //tableEnv.toAppendStream(resultTable, Row.class).print("filterResult:");

        // 3.2 聚合统计
        // Table API
        // 输出结果里的false表示上一条保存的记录被删除，true则是新加入的数据
        Table aggTable = inputTable.groupBy("id").select("id, id.count as count, temp.avg as avgTemp");
        // 打印输出
        //tableEnv.toRetractStream(aggTable, Row.class).print("aggResult:");

        // 自定义SQL
        Table sqlAggTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp from inputTable group by id").filter("id === 'sensor_c'");
        // 打印输出
        tableEnv.toRetractStream(sqlAggTable, Row.class).print("sqlResult:");

        env.execute();
    }

}
